package cn.monkey.transport.netty;

import cn.monkey.transport.core.Outbound;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import reactor.core.publisher.Mono;

import java.util.concurrent.Flow;

public interface NettyOutbound extends Outbound {

    Channel channel();

    @Override
    default Mono<Void> write(Object o) {
        ChannelFuture channelFuture = this.channel().write(o);
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        }));
    }

    @Override
    default Mono<Void> flush() {
        this.channel().flush();
        return Mono.empty();
    }

    @Override
    default Mono<Void> writeAndFlush(Object o) {
        ChannelFuture channelFuture = this.channel().writeAndFlush(o);
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        }));
    }
}
